package app.kafka.consume;

import java.util.Arrays;
import java.util.Properties;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import app.htby.utils.KafkaConsumeUtils;
import app.htby.utils.RedisUtils;
import app.htby.utils.SpringContextUtil;
import app.redis.JedisDataPlusPool;
import app.redis.JedisLocalPool;
import app.redis.ReconnDataPlusRedis;
import redis.clients.jedis.Jedis;

/**
 * 分发系统 消费 web系统 的kafka
 * 
 * @author lfy.xys
 * @date 2018年5月31日
 *
 */
public class DisWebConsumer {

	private KafkaConsumer<String, String> kafkaConsumer = null;
	private static Logger logger = LoggerFactory.getLogger(DisWebConsumer.class);

	public void setConsumer() {
		Properties props = new Properties();
		// bootstrap.servers 和Producer一样，是指向Kafka集群的IP地址，以逗号分隔。
		props.put("bootstrap.servers", KafkaConsumeUtils.bootstrapServers);
		props.put("group.id", KafkaConsumeUtils.dis_group_id);
		props.put("enable.auto.commit", KafkaConsumeUtils.enable_auto_commit);
		props.put("auto.commit.interval.ms", KafkaConsumeUtils.auto_commit_interval_ms);
		props.put("auto.offset.reset", KafkaConsumeUtils.auto_offset_reset);
		props.put("session.timeout.ms", KafkaConsumeUtils.session_timeout_ms);
		// 发序列化。Consumer把来自Kafka集群的二进制消息反序列化为指定的类型。因本例中的Producer使用的是String类型，所以调用StringDeserializer来反序列化
		props.put("key.deserializer", KafkaConsumeUtils.key_deserializer);
		props.put("value.deserializer", KafkaConsumeUtils.value_deserializer);
		kafkaConsumer = new KafkaConsumer<String, String>(props);
	}

	/**
	 * 测试 消费topic
	 * 
	 * @author lfy.xys
	 * @date 2018年6月5日
	 *
	 * @param topic
	 */
	public void consumeTest(String topic) {
		if (kafkaConsumer == null) {
			this.setConsumer();
		}
		kafkaConsumer.subscribe(Arrays.asList(topic));
		while (KafkaConsumeUtils.IS_CONSUMER_DIS) {
			// Consumer调用poll方法来轮循Kafka集群的消息，其中的参数100是超时时间（Consumer等待直到Kafka集群中没有消息为止）：
			ConsumerRecords<String, String> records = kafkaConsumer.poll(100);
			// 查看服务是否运行
			for (ConsumerRecord<String, String> record : records) {
				// TODO 获取的IOTPBean，需要转换
				String msg = record.value();
				System.out.println("------");
				System.out.println(msg);
				System.out.println("------");
			}

		}
		if(!KafkaConsumeUtils.IS_CONSUMER_DIS) {
			kafkaConsumer.close();
		}
	}
	
	/**
	 * 消费topic
	 * 
	 * @author lfy.xys
	 * @date 2018年6月5日
	 *
	 * @param topic
	 */
	public void consume(String topic) {
		if (kafkaConsumer == null) {
			this.setConsumer();
		}
		// Consumer订阅了Topic的消息，
		kafkaConsumer.subscribe(Arrays.asList(topic));
		// 判断是否继续 消费
		logger.info("是否分发数据：{}",KafkaConsumeUtils.IS_CONSUMER_DIS);
		while (KafkaConsumeUtils.IS_CONSUMER_DIS) {
			// Consumer调用poll方法来轮循Kafka集群的消息，其中的参数100是超时时间（Consumer等待直到Kafka集群中没有消息为止）：
			ConsumerRecords<String, String> records = kafkaConsumer.poll(100);

			// 查看服务是否运行
			for (ConsumerRecord<String, String> record : records) {
				// TODO 获取的IOTPBean，需要转换
				String msg = record.value();
				String bean = "http://domain/service_status/service_id/stauts_cpu_storageused-storagetotal_memused-memtotal_time";

				// TODO 1、插入 大数据的 redis
				logger.warn("//TODO ----------插入 大数据的 redis");
				try {
					Jedis jedis = JedisDataPlusPool.getJedis();
					jedis.sadd(RedisUtils.disDataPlusKey, bean);
					JedisDataPlusPool.returnRes(jedis);

					// 不报错，则大数据redis连接成功
					logger.info("连接大数据redis，发送数据成功。");

				} catch (Exception e) {
					// 大数据redis连接失败
					logger.error("大数据redis连接失败！！！");
					// kafka 停止拉取 数据
					KafkaConsumeUtils.IS_CONSUMER_DIS = false;
					
					/**
					// 失败的数据写入 本地redis
					try {
						Jedis jedis = JedisLocalPool.getJedis();
						jedis.sadd(RedisUtils.localBadDataKey, bean);
						JedisLocalPool.returnRes(jedis);

						// 不报错，则写入成功
						logger.info("连接本地redis成功，已写入发送失败数据。");
					} catch (Exception e2) {
						logger.error("本地redis连接失败！！！");
					}
					*/
					
					ReconnDataPlusRedis.IS_flag = false;
					
				}

				// TODO 2、插入 本地的 redis,和插入大数据redis 模式类型一样
				logger.warn("//TODO ----------插入 本地的 redis,和插入大数据redis 模式类型一样");

				// logger.info("--------------- offset ={}", record.offset());
				logger.info("---------topic:{} -----------msg：{}", topic, msg);
				
			}
		}
		
		//需要主动关闭
		if(!KafkaConsumeUtils.IS_CONSUMER_DIS) {
			kafkaConsumer.close();
		}
		
		
		
		
	}

	public static void main(String[] args) throws Exception {
		testKafkaConsumerSync();
	}
	
	/**
	 * 测试 kafka消费者 不拉取，是否会释放。
	 * 结果：不拉取数据，则释放
	 * @author lfy.xys
	 * @date 2018年6月5日
	 *
	 * @throws Exception
	 */
	public static void testKafkaConsumerSync() throws Exception {
		new Thread(new Runnable() {
			@Override
			public void run() {
				DisWebConsumer dwc = new DisWebConsumer();
				dwc.consumeTest("topic_1");
			}
		}).start();

		Thread.sleep(10 * 1000);
		KafkaConsumeUtils.IS_CONSUMER_DIS = false;
		Thread.sleep(3 * 1000);
		KafkaConsumeUtils.IS_CONSUMER_DIS = true;
		
		new Thread(new Runnable() {
			@Override
			public void run() {
				DisWebConsumer dwc1 = new DisWebConsumer();
				dwc1.consumeTest("topic_1");
			}
		}).start();
	}

}
